// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: gekim

#include "coin2/exchange/coincheck/feed_v1/processor.h"

#include <string>
#include <utility>

#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/coincheck/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::coincheck::feed_v1::impl {

void CoincheckFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  const auto& data = packet.payload;

  rapidjson::Document document;

  document.Parse(data.c_str());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: "
            << GetParseError_En(document.GetParseError()) << packet.payload;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  if (document.IsArray()) {
    auto items = document.GetArray();

    if (items[0].IsArray()) {
      // trade
      if (!CheckChannelFilter("trade")) {
        return;
      }
      for (auto& item: items) {
        auto trade_info = item.GetArray();
        const auto& native_symbol = trade_info[2].GetString();
        if (!upd_mgr_.is_symbol_registered(native_symbol)) {
          return;
        }
        Trade* trade = upd_mgr_.GetTradeMutable();
        CoincheckFeedUpdateParser::ParseTrade2(trade_info, trade, packet.timestamp);
        if (trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(native_symbol, false);
        }
      }
    } else if (items.Size() == 2) {
      // orderbook
      if (!CheckChannelFilter("orderbook")) {
        return;
      }
      const auto& native_symbol = items[0].GetString();
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        if (book->BidSize() > 0 && book->AskSize() > 0) {
          bool book_valid = CoincheckFeedUpdateParser::ParseBookUpdate(
              book, items[1], packet.timestamp);
          if (book_valid) {
            upd_mgr_.PublishBook();
          } else {
            LOG(ERROR) << "[Coincheck Feed Processor] wrong snapshot: "
                      << packet.topic_name << ": " << packet.payload.data();
          }
        }
      }
    } else if (items.Size() == 5){
      // trade old format
      if (!CheckChannelFilter("trade")) {
        return;
      }
      const auto& native_symbol = items[1].GetString();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        return;
      }
      Trade* trade = upd_mgr_.GetTradeMutable();
      CoincheckFeedUpdateParser::ParseTrade(items, trade, packet.timestamp);
      if (trade->fill_qty > 0) {
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    }
  } else if (document.IsObject()) {
    auto url_str = std::string(document["url"].GetString());
    auto symbol_pos = url_str.rfind("pair=");
    if (symbol_pos != -1) {
      std::string native_symbol = url_str.substr(symbol_pos + 5);
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book != nullptr) {
        if (book->BidSize() == 0 || book->AskSize() == 0) {
          bool book_valid = CoincheckFeedUpdateParser::ParseSnapshot(
              book, document["data"], packet.timestamp);
          if (book_valid) {
            upd_mgr_.PublishBook();
          } else {
            LOG(ERROR) << "[Coincheck Feed Processor] wrong snapshot: "
                       << packet.topic_name << ": " << packet.payload.data();
          }
        }
      }
    }
  }
  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::coincheck::feed_v1::impl
